package delivery

import (
	"bytes"
	"encoding/json"
	"errors"
	"github.com/jmoiron/sqlx"
	"io/ioutil"
	"learning/gooop/saga/mqs/database"
	"learning/gooop/saga/mqs/logger"
	"learning/gooop/saga/mqs/models"
	"net/http"
	"time"
)

type tDeliveryWorker struct {
	info        *tWorkerInfo
	successRing iMsgHistoryRing
	dbSource    iMsgSource
	liveSource  iMsgSource
	msgHeap     iMsgHeap
}

func newDeliveryWorker(info *tWorkerInfo) *tDeliveryWorker {
	it := new(tDeliveryWorker)
	it.init(info)
	return it
}

// init: do initialization, and start initial load
func (me *tDeliveryWorker) init(info *tWorkerInfo) {
	me.info = info
	me.successRing = newMsgHistoryRing(gMaxHistoryLogs)

	me.dbSource = newDBMsgSource(info.ClientID, me.isExpired)
	me.liveSource = newLiveMsgSource(info.ClientID, me.isExpired)
	me.msgHeap = newMsgHeap()

	go me.beginInitialLoadFromDB()
}

// beginInitialLoadFromDB: initially, load queued msg from database
func (me *tDeliveryWorker) beginInitialLoadFromDB() {
	buf := [][]*models.QueuedMsg{nil}
	for !me.isExpired() {
		err := database.DB(func(db *sqlx.DB) error {
			e, rows := me.loadFromDB(db)
			if e != nil {
				return e
			}

			buf[0] = rows
			return nil
		})

		if err != nil {
			logger.Logf("tDeliveryWorker.initialLoadFromDB, clientID=%s, err=%s", me.info.ClientID, err.Error())
			time.Sleep(3 * time.Second)
		} else {
			me.afterInitialLoad(buf[0])
			return
		}
	}
}

// loadFromDB: load queued msg from database
func (me *tDeliveryWorker) loadFromDB(db *sqlx.DB) (error, []*models.QueuedMsg) {
	rows, err := db.Queryx(
		"select * from delivery_queue where ClientID=? order by CreateTime asc limit ?",
		me.info.ClientID,
		gInitialLoadRows,
	)
	if err != nil {
		return err, nil
	}

	defer rows.Close()
	msgList := []*models.QueuedMsg{}
	for rows.Next() {
		msg := &models.QueuedMsg{}
		err = rows.StructScan(msg)
		if err != nil {
			return err, nil
		}
		msgList = append(msgList, msg)
	}

	return nil, msgList
}

// afterInitialLoad: after initial load done, push msgs into heap, and start delivery loop
func (me *tDeliveryWorker) afterInitialLoad(msgList []*models.QueuedMsg) {
	logger.Logf("tDeliveryWorker.afterInitialLoad, clientID=%s, rows=%d", me.info.ClientID, len(msgList))
	for _, it := range msgList {
		me.msgHeap.Push(it)
	}

	go me.beginPollAndDeliver()
}

// beginPollAndDeliver: poll msg from heap, and then deliver it
func (me *tDeliveryWorker) beginPollAndDeliver() {
	for !me.isExpired() {
		select {
		case msg := <-me.dbSource.MsgChan():
			logger.Logf("tDeliveryWorker.beginPollAndDeliver, msg from db=%v", msg)
			me.msgHeap.Push(msg)
			break

		case msg := <-me.liveSource.MsgChan():
			logger.Logf("tDeliveryWorker.beginPollAndDeliver, msg from live=%v", msg)
			me.msgHeap.Push(msg)
			break
		}

		if me.msgHeap.IsEmpty() {
			continue
		}

		msg := me.msgHeap.Pop()
		if msg == nil {
			continue
		}

		switch msg.StatusFlag {
		case 0:
			// 未处理的消息
			me.handleUndeliveredMsg(msg)
			break

		case 1:
			// 处理中的消息
			me.handleDeliveringMsg(msg)
			break
		}
	}
}

// isExpired: is me expired?
func (me *tDeliveryWorker) isExpired() bool {
	return time.Now().UnixNano() >= me.info.ExpireTime
}

// handleUndeliveredMsg: if msg unhandled, then try to deliver it
func (me *tDeliveryWorker) handleUndeliveredMsg(msg *models.QueuedMsg) {
	err := database.DB(func(db *sqlx.DB) error {
		now := time.Now().UnixNano()
		r, e := db.Exec(
			"update delivery_queue set StatusFlag=1, UpdateTime=? where id=? and StatusFlag=0 and UpdateTime=?",
			now,
			msg.ID,
			msg.UpdateTime,
		)
		if e != nil {
			return e
		}

		rows, e := r.RowsAffected()
		if e != nil {
			return e
		}
		if rows != 1 {
			return gOneRowsErr
		}

		msg.StatusFlag = 1
		msg.UpdateTime = now

		return nil
	})

	if err != nil {
		logger.Logf("tDeliveryWorker.handleNewMsg, id=%v, msg=%s/%s, err=%s", me.info.ClientID, msg.GlobalID, msg.SubID, err.Error())
		return
	}

	if me.deliver(msg) {
		me.afterDeliverySuccess(msg)

	} else {
		me.afterDeliveryFailed(msg)
	}
}

// deliver: use http.Post function to delivery msg
func (me *tDeliveryWorker) deliver(msg *models.QueuedMsg) bool {
	if me.successRing.Has(msg.ID) {
		return true
	}

	t := &models.TxMsg{
		GlobalID:   msg.GlobalID,
		SubID:      msg.SubID,
		Topic:      msg.Topic,
		CreateTime: msg.CreateTime,
		Content:    msg.Content,
	}
	logger.Logf("tDeliveryWorker.deliver, begin, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)

	j, e := json.Marshal(t)
	if e != nil {
		logger.Logf("tDeliveryWorker.deliver, failed json.Marshal, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
		return false
	}

	r, e := http.Post(me.info.NotifyURL, "application/json;charset=utf-8", bytes.NewReader(j))
	if e != nil {
		logger.Logf("tDeliveryWorker.deliver, failed http.Post, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
		return false
	}

	defer r.Body.Close()
	rep, e := ioutil.ReadAll(r.Body)
	if e != nil {
		logger.Logf("tDeliveryWorker.deliver, failed ioutil.ReadAll, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
		return false
	}

	m := &models.OkMsg{}
	e = json.Unmarshal(rep, m)
	if e != nil {
		logger.Logf("tDeliveryWorker.deliver, failed json.Unmarshal, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
		return false
	}

	if m.OK {
		logger.Logf("tDeliveryWorker.deliver, OK, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
		return true

	} else {
		logger.Logf("tDeliveryWorker.deliver, failed OkMsg.OK, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
		return false
	}
}

// handleDeliveringMsg: if delivery timeout, then retry delivery
func (me *tDeliveryWorker) handleDeliveringMsg(msg *models.QueuedMsg) {
	now := time.Now().UnixNano()
	if msg.UpdateTime+gDeliveryTimeoutNanos > now {
		return
	}

	// delivery timeout
	me.afterDeliveryTimeout(msg)
}

// afterDeliverySuccess: if done, move msg to success queue
func (me *tDeliveryWorker) afterDeliverySuccess(msg *models.QueuedMsg) {
	if me.successRing.Has(msg.ID) {
		return
	}
	me.successRing.Push(msg)

	err := database.TX(func(db *sqlx.DB, tx *sqlx.Tx) error {
		r, e := db.Exec(
			"delete from delivery_queue where GetID=? and UpdateTime=? and StatusFlag=1",
			msg.ID,
			msg.UpdateTime,
		)
		if e != nil {
			return e
		}

		rows, e := r.RowsAffected()
		if e != nil {
			return e
		}
		if rows != 1 {
			return gOneRowsErr
		}

		r, e = db.Exec(
			"insert into success_queue (MsgID, ClientID, CreateTime) values(?, ?, ?)",
			msg.ID,
			msg.ClientID,
			time.Now().UnixNano(),
		)
		if e != nil {
			return e
		}

		rows, e = r.RowsAffected()
		if e != nil {
			return e
		}
		if rows != 1 {
			return gOneRowsErr
		}

		return nil
	})

	if err != nil {
		logger.Logf("tDeliveryWorker.afterDeliverySuccess, failed, id=%v, msg=%s/%s, err=%s", me.info.ClientID, msg.GlobalID, msg.SubID, err.Error())
	} else {
		logger.Logf("tDeliveryWorker.afterDeliverySuccess, done, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
	}
}

// afterDeliveryFailed: if failed, do nothing but just log it
func (me *tDeliveryWorker) afterDeliveryFailed(msg *models.QueuedMsg) {
	logger.Logf("tDeliveryWorker.afterDeliveryFailed, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
}

// afterDeliveryTimeout: if timeout, then reset status and retry
func (me *tDeliveryWorker) afterDeliveryTimeout(msg *models.QueuedMsg) {
	err := database.DB(func(db *sqlx.DB) error {
		r, e := db.Exec(
			"update delivery_queue set StatusFlag=0 where id=? and StatusFlag=1 and UpdateTime=?",
			msg.ID,
			msg.UpdateTime,
		)
		if e != nil {
			return e
		}

		rows, e := r.RowsAffected()
		if e != nil {
			return e
		}

		if rows != 1 {
			return gOneRowsErr
		}

		return nil
	})

	if err != nil {
		logger.Logf("tDeliveryWorker.afterDeliveryTimeout, failed, id=%v, msg=%s/%s, err=%s", me.info.ClientID, msg.GlobalID, msg.SubID, err.Error())
	} else {
		logger.Logf("tDeliveryWorker.afterDeliveryTimeout, done, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
	}
}

var gEmptyRowsErr = errors.New("empty rows")
var gOneRowsErr = errors.New("expecting one row affected")
var gDeliveryTimeoutNanos = int64(10 * (time.Second / time.Nanosecond))
var gInitialLoadRows = 100
var gMaxHistoryLogs = 128
